package ru.maksb.exchange.server;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ru.maksb.exchange.domain.ClientInfo;
import ru.maksb.exchange.exceptions.ExchangeException;
import ru.maksb.exchange.service.Authenticator;
import ru.maksb.exchange.service.ExchangeWorker;
import ru.maksb.exchange.service.HeartbeatSender;
import ru.maksb.exchange.service.MessageHandler;
import ru.maksb.exchange.utils.MessageSender;

public class ServerWorker implements Runnable {
    private final static Logger logger = LoggerFactory.getLogger(ServerWorker.class);
    private final static String BUY = "buy";
    private static final long EXCHANGE_HEART_BEATS_DELAY = 10 * 1000;
    private AtomicBoolean serverWorkerRunning = new AtomicBoolean(true);
    private ExchangeWorker exchange;
    private Socket socket;
    private BlockingQueue<String> messagesToSend = new LinkedBlockingDeque<String>();
    private ObjectOutputStream out;
    private ObjectInputStream in;
    private ClientInfo clientInfo;
    private ScheduledExecutorService heartBeatExecutorService;

    public ServerWorker(Socket socket, ExchangeWorker exchange) throws ExchangeException {
        logger.debug("Connection on port {}", socket.getLocalPort());
        this.socket = socket;
        this.exchange = exchange;
        try {
            out = new ObjectOutputStream(socket.getOutputStream());
        } catch (IOException e) {
            throw new ExchangeException("Cannot handle connection", e);
        }
        try {
            in = new ObjectInputStream(socket.getInputStream());
        } catch (IOException e) {
            throw new ExchangeException("Cannot handle connection", e);
        }
    }

    public void run() {
        try {
            try {
                clientInfo = Authenticator.authenticate(in, out);
            } catch (IOException e1) {
                logger.error("Error in authentication", e1);
                return;
            } catch (ClassNotFoundException e1) {
                logger.error("Error in authentication", e1);
                return;
            }
            if (clientInfo == null) {
                logger.debug("Client login failed. Shutdown worker");
                return;
            }
            
            exchange.addWorker(this);
            
            new Thread(new MessageSender(messagesToSend, out)).start();
            heartBeatExecutorService = Executors.newSingleThreadScheduledExecutor();
            heartBeatExecutorService.scheduleAtFixedRate(new HeartbeatSender(messagesToSend), 
                                              EXCHANGE_HEART_BEATS_DELAY, 
                                              EXCHANGE_HEART_BEATS_DELAY, 
                                              TimeUnit.MILLISECONDS); 
            MessageHandler messageHandler = new MessageHandler(messagesToSend, 
                                                               exchange.getRequests(),
                                                               clientInfo);
            String message = null;
            while (serverWorkerRunning.get()) {
                try {
                    message = (String) in.readObject();
                    messageHandler.handle(exchange.getRequests(), message, serverWorkerRunning);
                    if (BUY.equals(message)) {
                        stopWorker();
                    }
                } catch (IOException e) {
                    logger.error("IO Message read error", e);
                    return;
                } catch (ClassNotFoundException e) {
                    logger.error("Rare message read error", e);
                } 
            }
        } finally {
            stopWorker();
        }
    }
    
    public void stopWorker() {
        logger.info("stopping worker for client {}", clientInfo.getName());
        exchange.closeWorker(clientInfo.getName());
        serverWorkerRunning.set(false);
        heartBeatExecutorService.shutdown();
        try {
            if (in != null) {
                in.close();
            }
        } catch (IOException e) {
            logger.error("stopping worker error", e);
        }
        try {
            if (out != null) {
                in.close();
            }
        } catch (IOException e) {
            logger.error("stopping worker error", e);
        }
        try {
            if (socket != null)
                socket.close();
        } catch (IOException e) {
            logger.error("stopping worker error", e);
        }
    }

    public ClientInfo getClientInfo() {
        return clientInfo;
    }
    
    public void sendMessage(String message) {
        try {
            messagesToSend.put(message);
        } catch (InterruptedException e) {
        }
    }
}
  